package com.eshipenet.shipnet.edgeboxextbuilder
import org.apache.flink.api.common.functions.AggregateFunction
import org.joda.time.DateTime

object EdgeBoxActionAggregate {
  type IN  = EdgeBoxAction
  type ACC = Map[String, Set[EdgeBox]]
  type OUT = Map[String, Set[EdgeBox]]
}

import EdgeBoxActionAggregate._

class EdgeBoxActionAggregate extends AggregateFunction[IN, ACC, OUT] {

  override def createAccumulator(): ACC = Map()

  override def add(value: IN, accumulator: ACC): ACC = value match {
    case EdgeBoxInsert(id, after, _) =>
      println("insert", after, accumulator)
      val modelId = after.modelId
      val sets = accumulator.get(modelId) match {
        case Some(s) => s + after
        case _       => Set(after)
      }
      val acc = accumulator.updated(modelId, sets)
      println("acc", acc)
      acc
    case EdgeBoxUpdate(id, before, after, raw) =>
      val bModelId = before.modelId
      val aModelId = after.modelId
      if (bModelId == aModelId) {
        val sets = accumulator.getOrElse(aModelId, Set()) + after
        accumulator.updated(aModelId, sets)
      } else {
        val acc = accumulator.get(bModelId) match {
          case Some(s) => accumulator.updated(bModelId, s - before)
          case _       => accumulator
        }
        val sets = acc.getOrElse(aModelId, Set()) + after
        acc.updated(aModelId, sets)
      }
    case EdgeBoxDelete(id, before, _) =>
      val modelId = before.modelId
      accumulator.get(modelId) match {
        case Some(sets) => accumulator.updated(modelId, sets - before)
        case _          => accumulator
      }
  }

  override def getResult(accumulator: ACC): OUT = accumulator

  override def merge(a: ACC, b: ACC): ACC = {
    println(s"a = $a, b = $b")
    val set = a.toSet ++ b.toSet
    set
      .groupBy(_._1)
      .map(v => (v._1, v._2.filter(_._2.nonEmpty).map(_._2)))
      .map(v => (v._1, combineAll(v._2)))
  }

  private def combineAll(set: Set[Set[EdgeBox]]): Set[EdgeBox] = {
    set.flatten.groupBy(_.id).map(kv => combineSet(kv._2)).toSet
  }

  private def combineSet(boxes: Set[EdgeBox]): EdgeBox = {
    implicit object DateTimeOrdering extends Ordering[EdgeBox] {
      override def compare(x: EdgeBox, y: EdgeBox): Int = ???
    }
    boxes.max[EdgeBox](DateTimeOrdering)
  }

}
